package org.apache.hadoop.mapred;

import java.util.*;
import java.io.*;
import java.lang.Exception;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.JsonMappingException;
import org.codehaus.jackson.map.ObjectMapper;

/**
 * NetworkSimulator
 * 
 * @author dongfei
 * 
 */
class NS implements NetworkSimulator {
	public int capacity = (int) (128 * 1024 * 1024); // the capacity is 128MBytes per second, 1Gbps
	private static final Log LOG = LogFactory.getLog(NS.class);
	boolean flag = true;
	long delay_time;
	
	public class Traffic {
		long startTime;
		long size;

		public Traffic(long startTime, long size) {
			this.startTime = startTime;
			this.size = size;
		}

		public void setSize(long size) {
			this.size = size;
		}
	}

	public long startTime = Long.MAX_VALUE;
	private Map<String, Traffic> maps = new HashMap<String, Traffic>();
	private ObjectMapper mapper;
	private JsonParser jsonParser;

	private String traceFile;

	public static String readFile(String fileName) {
		String output = "";

		File file = new File(fileName);
		try {
			BufferedReader input = new BufferedReader(new FileReader(file));
			StringBuffer buffer = new StringBuffer();
			String text;
			while ((text = input.readLine()) != null)
				buffer.append(text + "\n");

			output = buffer.toString();
		} catch (IOException ioException) {
			System.err.println("File Error!");
		}
		return output;
	}

	public void parse() throws JsonParseException, JsonMappingException,
			IOException {
		mapper = new ObjectMapper();
		mapper.configure(
				DeserializationConfig.Feature.CAN_OVERRIDE_ACCESS_MODIFIERS,
				true);
		File input = new File(this.traceFile);
		jsonParser = mapper.getJsonFactory().createJsonParser(input);
		HashMap<String, Object> obj = mapper.readValue(jsonParser,
				HashMap.class);

		ArrayList<HashMap<String, Object>> mapTasks = (ArrayList<HashMap<String, Object>>) obj
				.get("mapTasks");
		Integer inputBytes, outputBytes;
		String taskID = "";
		for (HashMap<String, Object> item : mapTasks) {
			taskID = (String) item.get("taskID");
			outputBytes = (Integer) item.get("outputBytes");
//			
			if (maps.get(taskID) != null) {
				LOG.debug("taskID: " + taskID + "\toutputBytes:" + outputBytes);
				maps.get(taskID).setSize(outputBytes);
			}
		}
	}

	public NS(String traceFile) {
		this.traceFile = traceFile;
	}

	public long getDelay(long now) {
		if(flag){
			flag = false;
			try {
				parse();
			} catch (Exception e) {
				System.err.println("Exception in parse!");
			}
			long delay = 0;
			Iterator<Traffic> it = maps.values().iterator();
			while (it.hasNext()) {
				double d_size = (double) it.next().size;
				delay +=  d_size*1000 ;
				
			}
			delay = (long) (delay / (double) capacity);
			
			long diff = (startTime + delay) - now;
			delay_time = Math.max(0, diff);
			LOG.debug("delay:"+delay+"Capacity:" + capacity);
			LOG.info("getDelay: " + delay_time + "\tnow:" + now + "\tnow-startTime:" + (now - startTime) + "\tdelay:" + delay);
			return delay_time;
		}else{
			LOG.info("getDelay: " + delay_time + "\tnow:" + now + "\tnow-startTime:" + (now - startTime));
			return delay_time;
		}
	}

	public void setTopology() {
	}
	
	@Override
	public void setMapInfo(String taskID, long finishTime) {
		// TODO Auto-generated method stub
		String task_name = "task_" + taskID.substring(8, taskID.length()-2);
		maps.put(task_name, new Traffic(finishTime,0));
		LOG.debug("task_name: " + task_name + "\finishTime:" + finishTime);
		if (finishTime < startTime){
			startTime = finishTime;
		}		
	}
}
